﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Actors;
using System.Threading.Tasks;

namespace ActorStressTests
{
    class PartitioningTests
    {
        // Partitioning greatly increases the number of actors on one IActorState.  We need to see if actor to actor method calls
        // will cause deadlocks.  We can simulate that with one normal actor calling a method on itself synchronously.
        public static void ActorToActorSynchronousCall(String fabricAddress, bool useGateway)
        {
            Console.WriteLine("Actor to Actor Synchronous blocking method call test");

            var actorCreator = new PartitionedActor("partitionedActor");
            using (var actorClient = actorCreator.DeployAndConnect(fabricAddress, useGateway, 3))
            {
                // Acquire the executing assembly and ship it to both actors
                var asm = typeof(PartitionedActor).Assembly;
                var name = asm.ManifestModule.Name;
                var location = asm.Location;
                var bytes = File.ReadAllBytes(location);

                actorClient.AddAssembly(name, bytes);

                Console.WriteLine("  Doing a method call that may deadlock...");
                actorClient.CallMethod("CallSameActorMethodSynchronously", new object[] { "fabric:/actor/adhoc/partitionedActor" });
            }
            Console.WriteLine("Actor to Actor synchronous call succeeded.");
        }

        // Mainly a performance test
        public static void ParallelLongRunningMethodCallsOnPartition(String fabricAddress, bool useGateway)
        {
            Console.WriteLine("Parallel long-running method calls on partitioned actor test");

            int numActors = 10; // Number of sub-actors to create, also serves as the number of threads
            int numIterations = 50;

            // Gather our assembly
            var asm = Assembly.GetExecutingAssembly();
            var name = asm.ManifestModule.Name;
            var location = asm.Location;
            var bytes = File.ReadAllBytes(location);

            // Create/connect to our sub-actors, shipping this assembly in the process
            FabricActorClient[] clients = new FabricActorClient[numActors];
            for (int i = 0; i < numActors; i++)
            {
                clients[i] = FabricActorClient.CreateOrConnectToActor(fabricAddress, "fabric:/actor/adhoc", new Uri("fabric:/actor/adhoc/perfPartition"),
                    "EmptyActorReplica", useGateway, numReplicas: 2, numPartitions: 2, partitionName: "actor" + i);
                clients[i].AddAssembly(name, bytes);
            }

            var nullParams = new object[0];
            int numErrors = 0;

            var sw = Stopwatch.StartNew();

            Console.WriteLine("Launching parallel runners to hammer partitioned actors");
            // Now launch some Threads to effect parallel hammering of partition
            Thread[] runners = new Thread[numActors];
            for (int i = 0; i < numActors; i++)
            {
                runners[i] = new Thread(obj =>
                    {
                        var myIndex = (int)obj;
                        var myClient = clients[myIndex];

                        try
                        {
                            for (int j = 0; j < numIterations; j++) myClient.CallMethod("LongRunningMethod", nullParams);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("Exception encountered by index {0}: {1}", myIndex, e);
                            Interlocked.Increment(ref numErrors);
                        }
                    });

                runners[i].Start(i);
            }

            Console.WriteLine("Waiting on runners...");
            for (int i = 0; i < numActors; i++) runners[i].Join();
            Console.WriteLine("Runners completed");
            sw.Stop();

            for (int i = 0; i < numActors; i++) clients[i].Dispose();
            Console.WriteLine("Disposed clients");

            if (numErrors > 0)
            {
                throw new Exception("Encountered errors.  See spew.");
            }
            else
            {
                Console.WriteLine("Successfully completed {0} iterations over {1} actors in {2} ms",
                    numIterations, numActors, sw.ElapsedMilliseconds);
            }
        }

        // A utility class for the PartitionCommunicationTortureTest.  Serves as an observer for "CallCount" events
        // being emitted from any particular actor.  Also allows test logic to wait until all expected
        // calls have been received.
        class CompletionObserver : IObserver<ActorEvent>
        {
            private TaskCompletionSource<int> m_tcs;
            private int m_expectedCompletions;
            private int m_latestCount = 0;
            private int m_index;

            public CompletionObserver(int index, int expectedCompletions)
            {
                m_index = index;
                m_expectedCompletions = expectedCompletions;
                m_tcs = new TaskCompletionSource<int>();
            }

            public void OnCompleted()
            {
            }

            public void OnError(Exception error)
            {
                throw new NotImplementedException();
            }

            public void OnNext(ActorEvent value)
            {
                int completions = value.GetTypedPayload<int>();
                string source = value.GetTypedPayloadQualifier<string>(0);

                m_latestCount = completions;
                //if ((completions % 50) == 0) Console.WriteLine("Actor {0} completion count: {1}", m_index, completions);
                //Console.WriteLine("Actor {0} completion count updated to {1}, source {2}", m_index, m_latestCount, source);
                if (completions < m_expectedCompletions) Console.Write("."); // signifies "working"
                else if (completions == m_expectedCompletions)
                {
                    Console.Write("!"); // signifies "done"
                    m_tcs.SetResult(completions);
                }
                else Console.Write("?"); // signifies "What?"
            }

            public void Wait()
            {
                m_tcs.Task.Wait();
            }
        }

        // Each of numActors actors makes an asynchronous call to each of the other numActors actors.
        // Meant to highlight the need for connection pooling and same-actor-short-circuiting of 
        // communications.
        public static void PartitionCommunicationTortureTest(String fabricAddress, bool useGateway, int numPartitions, int numActors)
        {
            Console.WriteLine("A2A Communication torture test, #partitions={0}, #actors={1} {2}", numPartitions, numActors, DateTime.Now);
            Console.WriteLine("  This creates a graph of actors, and the client subscribes to an event from");
            Console.WriteLine("  each one.  Then, each actor asynchronously calls a method on every other");
            Console.WriteLine("  actor, which calls another method to fire the event.");

            // Gather our assembly
            var asm = Assembly.GetExecutingAssembly();
            var name = asm.ManifestModule.Name;
            var location = asm.Location;
            var bytes = File.ReadAllBytes(location);

            // Create/connect to our sub-actors, ship this assembly to them, and hook up CompletionObservers
            FabricActorClient[] clients = new FabricActorClient[numActors];
            CompletionObserver[] observers = new CompletionObserver[numActors];
            var partitionedActorName = "fabric:/actor/adhoc/commTestPartition";
            for (int i = 0; i < numActors; i++)
            {
                clients[i] = FabricActorClient.CreateOrConnectToActor(fabricAddress, "fabric:/actor/adhoc", new Uri(partitionedActorName),
                    "EmptyActorReplica", useGateway, numReplicas: 2, numPartitions: numPartitions, partitionName: "actor" + i);
                clients[i].AddAssembly(name, bytes);
                observers[i] = new CompletionObserver(i, numActors - 1);
                clients[i].Subscribe("CallCount", observers[i]);
            }

            var sw = Stopwatch.StartNew();

            // Now kick off the communications on each actor
            // Do it asynchronously so that we don't get stuck on any one actor.
            List<Task> initializationTasks = new List<Task>();
            for (int i = 0; i < numActors; i++)
                initializationTasks.Add(clients[i].CallMethodAsync<object>("InitiateActorCalls", new object[] { numActors, i, partitionedActorName }));
            Task.WaitAll(initializationTasks.ToArray()); // will throw if any of the calls throws an exception.

            // (Intermediate output spew emitted by CompletionObservers)

            // Now wait for them to complete
            for (int i = 0; i < numActors; i++)
                observers[i].Wait();

            sw.Stop();

            Console.WriteLine("\n\nCompleted {0} x {1} A2A calls in {2} ms", numActors, numActors - 1, sw.ElapsedMilliseconds);

            // Now dispose the clients and destroy the partitioned actor
            Console.WriteLine("Disposing of partitioned actor state...  {0}", DateTime.Now);
            sw.Restart();
            for (int i = 0; i < numActors; i++)
            {
                clients[i].DeleteActor();
                clients[i].Dispose();
            }
            Console.WriteLine("Deleting the entire actor (the service instance).");
            FabricActorClient.DeleteActor(fabricAddress, partitionedActorName, true);
            sw.Stop();
            Console.WriteLine("Done disposing state.  Time for disposal: {0} ms", sw.ElapsedMilliseconds);
        }

    }

    public class PartitionedActor : FabricAdHocActorBase
    {
        public PartitionedActor(String name) : base(name)
        {
        }

        [ActorMethod]
        public static Object CallSameActorMethodSynchronously(IActorState state, Object[] parameters)
        {
            String actorName = (String)parameters[0];
            IActorProxy actorProxy = state.GetActorProxy(actorName);
            IActorPromise promise = actorProxy.Request("MethodCalledSynchronously", new Object[] { "Hi" });
            // Waiting here may cause a deadlock.  If the actor is located in the same partition as this one,
            // then this method and the target method being called will both be trying to take the same lock.
            promise.Wait();
            String ret = promise.GetTypedResult<String>();
            if (!String.Equals(ret, "Hi"))
                throw new Exception(String.Format("We called MethodCalledSynchronously passing it one string, but didn't get that string back as its return value!  Got: {0}", ret));

            return "CallSameActorMethodSynchronously succeeded!";
        }

        [ActorMethod]
        public static object LongRunningMethod(IActorState state, object[] parameters)
        {
            // Change some state to force the surrounding runtime logic to wait on the flush
            object temp;
            int count = 0;
            if (state.TryGet("_timeWastingCount", out temp))
                count = (int)temp;
            count++;
            state.Set("_timeWastingCount", count);

            // Now do something that takes a non-trivial amount of time
            Thread.Sleep(20);

            return count;
        }

        [ActorMethod]
        public static Object MethodCalledSynchronously(IActorState state, Object[] parameters)
        {
            return parameters[0];
        }

        // Initializes some variables and initiates the first of N calls of A2AMethodCall on other actors in partitioned actor
        [ActorMethod]
        public static object InitiateActorCalls(IActorState state, object[] parameters)
        {
            var actorCount = (int)parameters[0];
            var myIndex = (int)parameters[1];
            var partitionedActorName = (string)parameters[2];
            Random rnd = new Random(myIndex);
            var loopOffset = rnd.Next(0, actorCount); // Serves as randomizer to avoid actor access contention
            var loopIndex = (loopOffset == myIndex) ? 1 : 0; // Goes from 0 thru actorCount-1

            Console.WriteLine("Actor {0}: loopOffset = {1}", myIndex, loopOffset);

            // Save some important state
            state.Set("_actorCount", actorCount);
            state.Set("_myIndex", myIndex);
            state.Set("_partitionedActorName", partitionedActorName);
            state.Set("_loopIndex", loopIndex);
            state.Set("_loopOffset", loopOffset);

            // Kick off our first call
            var actualIndex = (loopIndex + loopOffset) % actorCount;
            var partitionKeyName = "actor" + actualIndex;
#if SYNC_PROXY
            var proxy = state.GetActorProxy(partitionedActorName, partitionKeyName);
            var promise = proxy.Request("A2AMethodCall", new object[] {"actor"+myIndex});
            promise.ContinueWith("ActorCallCompleted");
#else
            var proxyPromise = state.GetActorProxyAsync(partitionedActorName, partitionKeyName);
            proxyPromise.ContinueWith("GetProxyContinuation");
#endif
            //Console.WriteLine("actor {0} requesting proxy for actor {1}", myIndex, actualIndex);

            return null;
        }

        [ActorMethod]
        public static object GetProxyContinuation(IActorState state, object[] parameters)
        {
            var proxyPromise = (IActorPromise)parameters[0];
            var proxy = proxyPromise.GetTypedResult<IActorProxy>();
            var myIndex = (int)state.Get("_myIndex");
            var requestPromise = proxy.Request("A2AMethodCall", new object[] { "actor" + myIndex });
            requestPromise.ContinueWith("ActorCallCompleted");
            //Console.WriteLine("actor {0} calling A2AMethodCall", myIndex); 

            return null;
        }

        // Gets called as a continuation whenever an A2AMethodCall call to another actor completes
        [ActorMethod]
        public static object ActorCallCompleted(IActorState state, object[] parameters)
        {
            var result = (IActorPromise)parameters[0];  //ignore for now

            // Retrieve our state
            var actorCount = (int) state.Get("_actorCount");
            var myIndex = (int) state.Get("_myIndex");
            var partitionedActorName = (string) state.Get("_partitionedActorName");
            var loopIndex = (int) state.Get("_loopIndex");
            var loopOffset = (int)state.Get("_loopOffset");

            // Increment our loop index
            loopIndex++;
            int actualIndex;
            // Make sure that we don't call ourselves.
            while ( (actualIndex = ((loopIndex + loopOffset) % actorCount)) == myIndex) loopIndex++;
            if (loopIndex < actorCount)
            {
                // Update the loopIndex value in our state
                state.Set("_loopIndex", loopIndex);

                // Launch the next method call
                var partitionKeyName = "actor" + actualIndex;
#if SYNC_PROXY
                var proxy = state.GetActorProxy(partitionedActorName, partitionKeyName);
                var promise = proxy.Request("A2AMethodCall", new object[] { "actor" + myIndex });
                promise.ContinueWith("ActorCallCompleted");
#else
                var proxyPromise = state.GetActorProxyAsync(partitionedActorName, partitionKeyName);
                proxyPromise.ContinueWith("GetProxyContinuation");
#endif
                //Console.WriteLine("Actor {0} requesting proxy for actor {1}", myIndex, actualIndex);
            }

            return null;
        }

        // Basically just publishes the number of calls that have been made to this actor
        [ActorMethod]
        public static object A2AMethodCall(IActorState state, object[] parameters)
        {
            var source = (string) parameters[0];
            int numCalls = 0;
            object temp;
            if (state.TryGet("_numCalls", out temp)) numCalls = (int)temp;

            numCalls++;
            state.Set("_numCalls", numCalls);

            state.Publish("CallCount", numCalls,source);

            return numCalls; // just for the heck of it
        }


    }
}
